package com.mate.cloud;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @description:    RocketMQ 消费者
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/7/2 18:52
 * @updateUser:     MI
 * @updateTime:     2025/7/2 18:52
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class RocketMqConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer=
                new DefaultMQPushConsumer("mate_consumer_group");
        consumer.setNamesrvAddr("192.168.110.17:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("mate_test_topic","*");

        // 按顺序消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {

                MessageExt  messageExt=list.get(0);
                //TODO  --
                // Throw Exceptio
                // 重新发送该消息
                // DLQ（通用设计）
                if(messageExt.getReconsumeTimes()==3){  //消息重发了三次
                    //持久化 消息记录表
                    return ConsumeOrderlyStatus.SUCCESS; //签收
                }
                return ConsumeOrderlyStatus.SUCCESS; //签收
            }
        });

        consumer.start();

    }
}
